package com.lyx.question.mq;

import com.alibaba.fastjson.JSON;
import com.lyx.common.constant.RocketMQConstant;
import com.lyx.common.result.R;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;
import java.time.Instant;

/**
 * @author 14667
 * @date 2022/8/2 11:46
 */
@RestController()
public class Product {
    @Value("${rocket.namesrvAddr}")
    private String namesrvAddr;

    @Value("${rocket.topic}")
    private String topic ;

    @Value("${rocket.group}")
    private String group;

    @Value("${rocket.tag}")
    private String tag;



    @GetMapping("/scheuled")
    @ResponseBody
    public R schedule(@RequestParam Integer paperId,@RequestParam Integer userId,@RequestParam long delayTime) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {

        PaperMessage paperMessage = new PaperMessage();
        paperMessage.setPaperId(paperId);
        paperMessage.setUserId(userId);
        String message = JSON.toJSONString(paperMessage);
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(group,true);
        //元数据连接地址
        defaultMQProducer.setNamesrvAddr(namesrvAddr);
        defaultMQProducer.start();
        //定时消息投递时间
        final long deliverTimestamp = Instant.now().plusSeconds(delayTime).toEpochMilli();
        //创建消息对象
        Message msg = new Message(topic, tag, "KEY", message.getBytes(StandardCharsets.UTF_8));
        //设置消息定时投递的时间戳属性
        msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(deliverTimestamp));
        //发送消息
        SendResult sendResult = defaultMQProducer.send(msg);
        System.out.println("发送定时消息成功");
        defaultMQProducer.shutdown();
        return R.ok();
    }

    public void scheduleSendMessage(String message,long delayTime) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(group,true);
        //元数据连接地址
        defaultMQProducer.setNamesrvAddr(namesrvAddr);
        defaultMQProducer.start();
        //定时消息投递时间
        final long deliverTimestamp = Instant.now().plusSeconds(delayTime).toEpochMilli();
        //创建消息对象
        Message msg = new Message(topic, tag, "KEY", message.getBytes(StandardCharsets.UTF_8));
        //设置消息定时投递的时间戳属性
        msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(deliverTimestamp));
        //发送消息
        SendResult sendResult = defaultMQProducer.send(msg);
        System.out.println("发送定时消息成功");
        defaultMQProducer.shutdown();
    }
}
